/**  
 * @Title: DataXComp.java
 * @Description: TODO(描述)
 * @author liuyi
 * @date 2023-04-13 16:21:25 
 */
package com.fsxgt.datagrid.component.handler.impl;

import static com.fsxgt.datagrid.datax.tool.DataXConstant.DEFAULT_DATAX_PY;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson2.JSON;
import com.fsxgt.datagrid.component.bo.CmpSyncOfflineBO;
import com.fsxgt.datagrid.component.dao.CmpSyncOfflineMapper;
import com.fsxgt.datagrid.core.utils.LogHelpper;
import com.fsxgt.datagrid.datax.DataxJsonHelper;
import com.fsxgt.datagrid.datax.tool.AnalysisStatistics;
import com.fsxgt.datagrid.datax.tool.DataxSecretUtil;
import com.fsxgt.datagrid.datax.tool.LogStatistics;
import com.fsxgt.datagrid.datax.tool.ProcessUtil;
import com.fsxgt.datagrid.ds.component.QueryComponent;
import com.fsxgt.datagrid.ds.exception.ParamException;
import com.fsxgt.datagrid.ds.handler.DbHandler;
import com.fsxgt.datagrid.ds.utils.AESUtil;
import com.fsxgt.datagrid.ds.utils.JdbcConstants;
import com.fsxgt.datagrid.ds.utils.ParamUtil;
import com.fsxgt.datagrid.flow.exception.FlowException;
import com.fsxgt.datagrid.flow.handler.FlowContext;
import com.fsxgt.datagrid.flow.handler.FlowReturn;
import com.fsxgt.datagrid.flow.handler.IComponentHandler;
import com.fsxgt.datax.DataxFast;
import com.fsxgt.datax.TaskExecutor;
import com.github.pagehelper.util.StringUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * @ClassName: DataxComponent
 * @Description: TODO(描述)
 * @author liuyi
 * @date 2023-04-13 16:21:25
 */
@Slf4j
@Component
public class CmpSyncOfflineHandler extends IComponentHandler {


	@Autowired
	private QueryComponent queryComponent;

	@Autowired
	private DataxJsonHelper dataxJsonHelper;

	@Value("${dataXHomePath:null}")
	private String dataXHomePath;

	@Autowired
	private JdbcTemplate jdbcTemplate;
	@Autowired
	private CmpSyncOfflineMapper cmpSyncOfflineMapper;

	private String[] buildDataXExecutorCmd(Object startId, Object endId, CmpSyncOfflineBO bo, String jsonFile, Map params) throws ParamException {
		
		String dataXPyPath = null;

		List<String> cmdArr = new ArrayList<>();
		cmdArr.add("python");
		if (StringUtils.isNotEmpty(dataXHomePath)) {
			dataXPyPath = dataXHomePath.contains("bin") ? dataXHomePath + DEFAULT_DATAX_PY
					: dataXHomePath + "bin" + File.separator + DEFAULT_DATAX_PY;
		}
		cmdArr.add(dataXPyPath);

		

		StringBuilder doc = new StringBuilder();
		
		if(bo.getToolType()!=null&&bo.getToolType().equals("datax-shell")) {
			
			String text=ParamUtil.transform(bo.getCmdParam(), params).getText();
			
			
			text=ParamUtil.replace(text, "${endId}", ParamUtil.getParamValue(null,endId));
			text=ParamUtil.replace(text, "${startId}", ParamUtil.getParamValue(null,startId));
			
			doc.append(text);
			
			if (StringUtils.isNotBlank(doc)) {
				cmdArr.add(doc.toString());
			}
			
			cmdArr.add(jsonFile);
			
			return cmdArr.toArray(new String[cmdArr.size()]);
			
		}else {
			
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

//			doc.append(JVM_CM).append(TRANSFORM_QUOTES).append(jvmParam).append(TRANSFORM_QUOTES);
			String param = "-p \"-DstartId=%s\" \"-DendId=%s\" ";

			if (StringUtils.isNotEmpty(bo.getSrcTableName())) {
				
				param = String.format(param, ParamUtil.getParamValue(endId.getClass(),startId), ParamUtil.getParamValue(endId.getClass(),endId));

			} else {
				param = String.format(param, ParamUtil.getParamValue(null,startId) , ParamUtil.getParamValue(null,endId));

			}

			doc.append(param);

			if (StringUtils.isNotBlank(doc)) {
//	            cmdArr.add(doc.toString().replaceAll(SPLIT_SPACE, TRANSFORM_SPLIT_SPACE));
				cmdArr.add(doc.toString());
			}
			cmdArr.add(jsonFile);

			return cmdArr.toArray(new String[cmdArr.size()]);
		}
		

	}

	private String buildDataxFile(CmpSyncOfflineBO bo, Map params) throws Exception {

		String jsonText=null;
		
		if(bo.getToolType()!=null&&bo.getToolType().equals("datax-shell")) {
			jsonText=bo.getJsonText();
			jsonText=DataxSecretUtil.dataxDeCrypt(jsonText);
		}else {
			jsonText=JSON.toJSONString(dataxJsonHelper.buildJSON(bo, params));
		}
		
		jsonText=ParamUtil.transform(jsonText, params).getText();
		
		jsonText=DataxSecretUtil.dataxDeCrypt(jsonText);
		

		File file = File.createTempFile("datax", ".json");
		FileUtils.writeStringToFile(file, jsonText, "utf-8");
		return file.getPath();
		

		

	}

	private Object convertValue(Object value, Long offset) {

		Object result = null;

		SimpleDateFormat[] sdfs = new SimpleDateFormat[] { new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"),
				new SimpleDateFormat("yyyy-MM-dd"), new SimpleDateFormat("yyyy-MM-dd HH") };

		if (value instanceof Date) {
			result = new Date(((Date) value).getTime() + offset);
		} else if (value instanceof LocalDateTime) {
			String srcStartId = ((LocalDateTime) value).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
			result = ((LocalDateTime) value).plusSeconds(offset / 1000);
		} else if (value instanceof String) {
			for (SimpleDateFormat sdf : sdfs) {
				try {
					Date date = sdf.parse((String) value);
					result = new Date(((Date) date).getTime() + offset);
					break;
				} catch (Exception e) {
					;
				}
			}
		} else {
			result = ((Number) value).doubleValue() + offset;
		}
		return result;
	}

	public static void main(String args[]) throws ParseException {

		System.out.println(Number.class.isAssignableFrom(new Float("11").getClass()) );

	}

	public String getSrcStartId(CmpSyncOfflineBO bo) {
		return cmpSyncOfflineMapper.get(bo.getFlowInfoId(), bo.getComponentSeq()).getSrcStartId();
	}

	public void callBack(CmpSyncOfflineBO bo, Object srcMaxValue) {

		if (srcMaxValue == null) {
			return;
		}

		String sql = "update cmp_sync_offline set src_start_id=? where flow_info_id=? and component_seq=?";

		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

		if (srcMaxValue instanceof Date) {
			jdbcTemplate.update(sql, sdf.format(srcMaxValue), bo.getFlowInfoId(), bo.getComponentSeq());
		} else if (srcMaxValue instanceof LocalDateTime) {
			String srcStartId = ((LocalDateTime) srcMaxValue)
					.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
			jdbcTemplate.update(sql, srcStartId, bo.getFlowInfoId(), bo.getComponentSeq());
		} else {
			if(srcMaxValue!=null) {
				jdbcTemplate.update(sql, srcMaxValue.toString(), bo.getFlowInfoId(), bo.getComponentSeq());
			}
		}
	}

	public FlowReturn handler(final Object config, FlowContext context) throws Exception {
		

		CmpSyncOfflineBO bo = (CmpSyncOfflineBO) config;

		Map params = (Map) context.get(ParamUtil.PARAM_KEY);

		if (StringUtils.isNoneEmpty(bo.getSrcSchema()) && StringUtils.isNoneEmpty(bo.getSrcTableName())
				&& bo.getSrcTableName().indexOf(".") < 0) {
			bo.setSrcTableName(bo.getSrcSchema() + "." + bo.getSrcTableName());
		}

		if (StringUtils.isNoneEmpty(bo.getTargetSchema()) && StringUtils.isNoneEmpty(bo.getTargetTableName())
				&& bo.getTargetTableName().indexOf(".") < 0) {
			bo.setTargetTableName(bo.getTargetSchema() + "." + bo.getTargetTableName());
		}

		bo.setSrcStartId(getSrcStartId(bo));

		DbHandler dbHandler = queryComponent.getHandler(bo.getSrcDatasourceId());
		log.info(String.format("获取target表最大%s:%s", bo.getSrcIdExpression(), bo.getSrcStartId()));

		Object srcMaxValue = null;

		if (StringUtils.isNotEmpty(bo.getSrcTableName())) {
			
			String where = ParamUtil.transform(bo.getSrcWhereParam(), params).getText();
			srcMaxValue = dbHandler.getMaxIdVal(bo.getSrcTableName(), bo.getSrcIdExpression(), where);
			log.info(String.format("获取src表最大%s:%s", bo.getSrcIdExpression(), srcMaxValue));
			if (srcMaxValue == null) {
				log.info("src表无数据");
				return new FlowReturn();
			}
		} else if (StringUtil.isNotEmpty(bo.getSrcIdExpression())) {

			String text = bo.getSrcSqlText();

			text = ParamUtil.transform(text, params).getText();

			String sql = dbHandler.getMaxSqlForExp(text, bo.getSrcIdExpression());

			sql = ParamUtil.removeParam(sql, " 1=1 ");

			log.info(String.format("执行sql:%s", sql));

			srcMaxValue = dbHandler.queryForObject(sql);

			log.info(String.format("获取src表最大%s:%s", bo.getSrcIdExpression(), srcMaxValue));
		}
		

		if(bo.getToolType()!=null&&bo.getToolType().equals("datax-fast")) {
			
			//规避clickhouse一致性问题
			if(dbHandler.getDatasourceType().equals(JdbcConstants.CLICKHOUSE)) {
				Thread.sleep(3000);
			}
			
			String text = JSON.toJSONString(dataxJsonHelper.buildJSON(bo, params));	
			
			if (StringUtils.isNotEmpty(bo.getSrcTableName())) {
				text=ParamUtil.replace(text, "${endId}", ParamUtil.getParamValue(srcMaxValue.getClass(),srcMaxValue));
				text=ParamUtil.replace(text, "${startId}", ParamUtil.getParamValue(srcMaxValue.getClass(),bo.getSrcStartId()));
			}else {
				text=ParamUtil.replace(text, "${endId}", ParamUtil.getParamValue(null,srcMaxValue));
				text=ParamUtil.replace(text, "${startId}", ParamUtil.getParamValue(null,bo.getSrcStartId()));
			}
			
			text=ParamUtil.transform(text, params).getText();

			TaskExecutor taskExecutor = DataxFast.getTaskExecutor(text);
			context.put("taskExecutor"+bo.getId(), taskExecutor);
			
			taskExecutor.doStart();
			
			callBack(bo, srcMaxValue);
	
		}else {
			
			String filePath = buildDataxFile(bo, params);

			String[] cmd = buildDataXExecutorCmd(bo.getSrcStartId(), srcMaxValue, bo, filePath,params);

			String cmdStr = "";

			for (String s : cmd) {
				cmdStr += s + " ";
			}

			log.info("---------dataxCmd:" + cmdStr);

			Thread errThread = null;
			LogStatistics logStatistics = null;
			int exitValue = -1;

			try {

				final Process process = Runtime.getRuntime().exec(cmd);
				String prcsId = ProcessUtil.getProcessId(process);

				context.put("dataxProcessId"+bo.getId(), prcsId);

				log.info("DataX process id: " + prcsId);

				String logFileName = LogHelpper.getLogFileName();
				// log-thread
				Thread futureThread = null;
				FutureTask<LogStatistics> futureTask = new FutureTask<>(new Callable() {
					@Override
					public Object call() throws Exception {
						// TODO Auto-generated method stub
						LogHelpper.setLogFileName(logFileName);
						return AnalysisStatistics.analysisStatisticsLog(new BufferedInputStream(process.getInputStream()));
					}
				});

				futureThread = new Thread(futureTask);
				futureThread.start();

				errThread = new Thread(() -> {
					try {
						LogHelpper.setLogFileName(logFileName);
						AnalysisStatistics.analysisStatisticsLog(new BufferedInputStream(process.getErrorStream()));
					} catch (IOException e) {
						log.error("", e);
					}
				});

				logStatistics = futureTask.get();
				errThread.start();
				// process-wait
				exitValue = process.waitFor(); // exit code: 0=success, 1=error
				// log-thread join
				errThread.join();

				if (exitValue != 0) {
					throw new FlowException("datax线程运行失败");
				}

				callBack(bo, srcMaxValue);

			} finally {
				if (errThread != null && errThread.isAlive()) {
					errThread.interrupt();
				}
				FileUtils.forceDeleteOnExit(new File(filePath));

			}
		}


		return new FlowReturn();

	}

	@Override
	public String getComponentType() {
		// TODO Auto-generated method stub
		return "Sync Offline";
	}

	@Override
	public void stopHandler(Object config, FlowContext context) throws Exception {
		// TODO Auto-generated method stub
		
		CmpSyncOfflineBO bo = (CmpSyncOfflineBO) config;
		
		if(context.get("dataxProcessId"+bo.getId())!=null) {
			String pid = context.get("dataxProcessId"+bo.getId()).toString();
			log.info("终止线程: " + pid);
			ProcessUtil.killProcessByPid(pid);
			log.info("终止线程成功: " + pid);
		}
		
		if(context.get("taskExecutor"+bo.getId())!=null) {
			TaskExecutor taskExecutor= (TaskExecutor) context.get("taskExecutor"+bo.getId());
			taskExecutor.doStop();
		}
		
	}

}
